Heron 论文翻译及理解

背景介绍：

Heron是号称Twitter流数据处理的新一代实现，是StormV2。我们首先回顾一下Storm系统的问题

1. worker日志混乱，如果一个bolt日志过大，会冲掉其他bolt的日志
2. worker之间因为没有资源隔离，因此会出现不确定的worker间相互影响
3. Nimbus单点故障
4. 背压(Back-Pressure)问题:receiver无法处理消息时，sender会丢弃消息。这种情况下如果没有ack将无法得知是否所有消息是否被处理，在最极端的情况下，系统会不同的消耗资源但得不到任何结果。

针对这些问题，我们实现了Heron：

数据模型：支持At Most once 以及At least once

工作模型：

调度器（基于Mesos的Aurora）：

用于对每个topology使用的资源进行分配。

Topology Master （TM）

在调度器分配资源后，topology内部有一个类似选主的过程，主通过在zk上锁定一个公认的node以标示自己为主。同时，TopologyMaster也作为监控中心收集topology内部运行状态。但不作为数据流的中心，因此topology Master不会成为瓶颈。

Stream Manager（SM） 以及Heron Instance（HI）

被分配到同一个topology中的SM之间建立全连接，即O（N*N）的连接，可见单个topology规模越大，SM建立的连接数量也会成倍增加。

HI则为分配到SM内部的处理单元，HI通过SM的信息通路进行通信。

Topology Back Pressure

背压机制主要用于流量控制，该机制保证了不同的组件可以以不同的速度处理事件。
举例而言：拓扑中有上游组件以及下游组件，当下游组件因为意外处理速度变慢时，上游组件依然以原速度发送事件，则会导致事件出现堆积。无论是中间丢弃事件还是把消息队列撑爆，显然都是我们不希望看到的情况。
这时候，背压机制的作用就是调节上游组件，让其减慢速度。

尝试1：TCP 背压

我们尝试在TCP连接的层面对该问题进行调整。SM之间通过TCP进行连接，二者之间都有Sending Buffer以及Receiver Buffer。当receiver处理速度变慢时，Receiver Buffer会随之变满，而后SendingBuffer也会慢，这时候上游组件是可以察觉到这种情况的。而后可以进行其他措施进行处理，如发送给其他的SM，或者调整自己的处理速度，直到下游SM速度赶上来位置。

这种方式实现简单，但实际效果并不理想。Because multiple logical channels (between HIs) are overlaid on top of the physical connections between SMs. 因为HI之间的多个逻辑连接都建立在SM的连接至上，该策略不仅降低了上游SM的发送速度，同时也降低了下游被背压的HI的处理速度。

其结果是，任何处理速度下降引起的TCP背压都会导致一连串的连锁反应导致系统处于长期的动荡调整阶段。

尝试2: Spout 背压：

这种方式同样结合了Sending Buffer以及Receiver Buffer进行。当SM意识到自己内部的某个HI处理速度变慢时，SM找到对应HI的spout并停止从该Spout中读取数据。而该Spout的发送缓存也因此会被填满并最终导致阻塞。与次同时SM会发送“开始背压消息”（Start BackPressure Message）到该topology的其他SM中。收到“开始背压消息”的SM则全部停止从自己上游获取消息的举措。
即：让其他所有的组件全部停下等待速度慢的组件消化完自己Receiver Buffer中的消息之后大家在一起继续。
当缓慢的HI速度赶上来之后，SM再次发送“停止背压消息”，收到消息的SM从新进入正常工作状态。

分析：该方式就是让其他所有快的都等慢的。同时隐含着系统被大量 stop start back pressure冲毁的风险。但其好处也是显而易见的，无论topology的层次有多深，该方式都能够保证足够低的处理延时。

尝试3：

层层背压：没懂，还好 也没用

实现：

我们最终实现的是Spout背压机制。该机制在实践中工作良好且可以轻易得出哪里是问题的root cause。
在实现上，每个socket连接都具有两个界限值，即高水位界限值以及低水位界限值。当receiver buffer到达高水位界限值时触发背压策略，直到Buffer回复到低水位时终止背压。设计成这种策略主要是防止系统反复震荡。
因为有了背压策略，所以系统不会在丢弃tuple，因此系统只有在出现组件故障时才回丢弃tuple，这使得tuple的丢弃更具有确定性。

当此时注意到，一旦进入背压模式，整个系统的处理速度与系统中最慢的组件速度相当。

HI 的实现

HI 抛弃Storm原有的多个Spout或者bolt可以共享一个JVM的思路，每个HI单独使用一个JVM，因此这样可以方便的调试各个组件的性能。

与此同时，因为传输事件的任务交由SM完成，因此对于我们而言，HI可以是以任意语言编写的代码。

对于HI的实现，我们主要有两种设计思路：单线程的实现以及多线程的实现。下面我们将分述两种实现。

单线程实现

在单线程的实现中，线程维护一个与其所属的SM的TCP连接。一旦有事件到达，users logic代码会被调用，如果user的代码会生成tuple，则该线程同样负责将该tuple传递给所属的SM。

该代码虽然简单，但存在诸多问题，最主要原因是usercode可能会各种原因被block

1. 为了减速小睡一会
2. 调用系统IO，TCP连接，文件操作等需要上下文切换频繁的工作
3. 调用一些同步线程（synchronization）方法

这些block最关键的问题是会导致一些定时操作被耽误，例如性能状态监控。因为这些block时间不够确定，因此可能导致性能状态metric收集不及时，让master以为HI处于非良好状态。

双线程实现

如名所属，该方式下每个HI通过两个thread实现，一个GateWay Thread一个Task Execution Thread。

如图所示GateWay Thread负责与SM以及状态监控的MM进行通信，收发消息。Task Excurtion Thread 负责执行用户的逻辑代码。

两个线程之前通过三条单向有界队列（unidirection Queue）进行连接。当Data in 队列满时，GateWay不再从SM中读取消息，与此同时触发其所在SM的背压机制。同样，当dataout队列满时（由此可知队列buffer应该都由gate thread维护），GateWay同样认为此时TaskExcution线程不能接收更多的消息从而定制发送消息。与此同时TaskE线程也不再向dateout队列发送消息。

对TaskE线程而言，如果他是spout，则该线程不停地执行nextTuple方法并向data out队列推送消息。如果是bolt，则每个从datain 进入的消息都会触发 execute方法。

GC问题：
当我们使用topology的有界队列执行大型topology时，我们进场遇到GC问题。正当一切都正常的时候，一个网络中断（network outage）发生了，因此SM无法消费GateWay发出的消息因此dataout队列会发生堆积，且因为这些事件需要被处理，因此无法被GC回收。因此很容易导致HI达到其内存极限。
此时，当网络恢复时，Gateway从SM读入新的tuple的同时也向SM发送堆积在outqueue中的消息。如果GateWay读入消息在前，新object的生成占用更多内存从而触发GC，而之前outqueue因为堆积占用了所有内存，因此会导致更严重的性能下降(Performance degradation)。

为了解决此类GC问题，我们主要的思路是定期检查datain queue 以及dataout queue的大小，并动态更改其capacity。如果这些队列的capacity超过了某个设定的极限，则系统以折半的方式减小其capacity。
（这里没有说具体减小的方式，个人认为对于缩减datain queue，就是不在从SM中读取数据，知道datain queue中的tuple被消耗到一半之后完成一次折半capacity的过程。同样，对于out queue，也是先停in queue，然后直到out queue中消息被发出去一半之后再继续）。
这种折半减少capacity的方式被定时的调用直到：queue的capacity到达一个稳定的状况或者queue的大小减到0。如果queue的大小减到0，显然对于大多数的情况下，没有tuple可以被生产以及消费，（系统进入一个类似无状态的过程），此时进行GC会变得容易得多。

相反如果queue中的tuple数量小于设置值，系统会（通过GateWay从SM中读入更多的数据）增加queue中tuple的数量，知道到达极限值或者设置值。

MetricManager

MM负责收集并发送所有的状态。包括系统状态以及用户定义的的状态。

开始过程以及异常处理

• 启动过程：

  在提交一个topology之后调度器（Aurora）allocate必须的资源并在这些机器上启动topology container。Topology container中选出Topology Master，并在zk上通过锁住一个公认的节点宣布其为Master。与此同时，所有的SM（Stream Master）通过zk发现TM并与之进行定期心跳。
  当所有SM与TM全部连接完毕(不需要两两互联)，TM开始运行一个部署算法将topology中不同的components分配到不同的container中。在我们的术语中，我们称之为physical plan。完成Physical Plan之后SM与TM通信获得其按照Plan应该与之互联的SM。此时，SM已经完成了一个完全互通的网络。HI启动并按照Plan获得其需要执行的任务，至此，整个topoloogy构建完毕，数据按照规定开始流动。
  为了保险起见，TM可以Plan写入ZK以防止TM单点故障。

• 异常处理：

  在Topology执行的过程中，有几种故障情况可能引起topology的整个执行过程。

• TM Die：

  当TM挂掉时，container会重启TM进程，TM进程则会自动从ZK回复状态。
  （个人问题1： TM实时将状态写入ZK？包括哪些状态？）
  如果存在Standby TM，则主备切换。与TM建立Channel的SM全部与之前的备建立Channel。

• SM Die：

  与TM类似，SM死掉时，它会被其所在的Container重启，被重启之后的SM首先联系ZK找TM，之后与TM通信下载Plan，以查看是否有任务变化。同时，与该死掉的SM互联的SM在与其失去连接后，同样连接TM下载新Plan查看是否应该找个新的SM连接还是等待其重启。

• HI Die：

  HI重启之后则直接与自己的SM连接获取plan并重新工作。

译者补充:

这里讲的比较模糊，有几个问题没有讲清楚：

1. ZK里存得Plan是一个什么形式？
2. TM实时更新了哪些状态？
3. SM。HI挂了的时候，当前正在发送的消息和SM内部的消息是怎么处理的？

• 总结

1. cluster Manager（Nimbus）的任务被明确的分开了

2. 每个HI仅仅执行一个任务，所以debug log变得简单了

3. 因为设计将整个执行过程变得透明化，当topology变慢时，可以清晰的找到问题的根源。
4. 资源隔离
5. 每个topology都具有自己的TM
6. 背压机制

译者总结：

改变1：

对照 原来的storm

HI->woker

container->Supervisor。

而SM则是从原来的worker中提出来的通信组件。原来的模式中每个woker负责与其他worker通信，在我们实际使用中，一个woker里能开到100多个线程，其中发送接收都开了很多线程。每个worker都开这么多与其他机器的连接显然是很消耗资源而且很不可靠的方式。

而，参考现在的方式，SM被单独提出来负责所有通信，每个HI只需要跟SM建立本地连接。所以，此时，一旦出现单点故障，只需要SM负责通信上的问题即可。HI完全不受影响。而现在的storm立刻出现大量连接断开，甚至出现雪崩现象。

改变2：

Nimbus抽象到TM和schedule两部分，并且通过ZK维护整个Topology的状态。因为我们的使用规模依然较小，并没有遇到TM需要解决问题的痛点，不过多评论。